package com.ruyuan.engine.core;

import com.ruyuan.engine.datagen.RuleComplexParamDataGen;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.RuleComplexParam;
import com.ruyuan.engine.pojo.RuleResult;
import com.ruyuan.engine.router.Router;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于router查询
 * */
public class RuleEngineProcessDynamicByRouter extends KeyedProcessFunction<String, ClientLog, RuleResult> {
    protected static final Logger log = LoggerFactory.getLogger(RuleEngineProcessDynamicByRouter.class);
    ListState<ClientLog> clientLogListState;
    RuleComplexParam ruleComplexParam;
    Router router;


    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<ClientLog> descriptor = new ListStateDescriptor<ClientLog>("clientLogListState",ClientLog.class);
        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.hours(6)).updateTtlOnCreateAndWrite().build();
        descriptor.enableTimeToLive(stateTtlConfig);
        clientLogListState = getRuntimeContext().getListState(descriptor);
        ruleComplexParam = RuleComplexParamDataGen.mockData();
        router = new Router(clientLogListState);
    }

    @Override
    public void processElement(ClientLog clientLog, Context ctx, Collector<RuleResult> out) throws Exception {
        clientLogListState.add(clientLog);
        if (ruleComplexParam != null) {
                if (!router.isMatchRule(clientLog,ruleComplexParam)) return;
                out.collect(new RuleResult(ruleComplexParam.getRuleId(),clientLog.getEventId(),System.currentTimeMillis()));
        }
    }
}
